package com.lagou.no4

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
object Staet {
    def main(args: Array[String]): Unit = {
        //获取流环境
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        //读取socket数据流并且分组封装为KV形式
        val kvDS: KeyedStream[(String, Int), String] = env.socketTextStream("node01", 9999)
          .map { line =>
              val arr = line.split(",")
              (arr(0).trim, arr(1).trim.toInt)
          }.keyBy(_._1)

        val res: DataStream[(String, Int)] = kvDS.map(
            new RichMapFunction[(String, Int), (String, Int)] {
                //声明一个最大值的状态变量
                var maxValueState: ValueState[Int] = _

                //获取上下文对象，获取状态
                override def open(parameters: Configuration): Unit = {
                    //状态描述器
                    val maxValue: ValueStateDescriptor[Int] = new ValueStateDescriptor[Int]("maxValue", classOf[Int])
                    //获取状态
                    maxValueState = getRuntimeContext.getState(maxValue)

                }

                //value是新来的数据
                override def map(value: (String, Int)): (String, Int) = {
                    //将较大的数更新进状态
                    maxValueState.update(Math.max(maxValueState.value(), value._2))
                    //返回对应的key的最大值
                    (value._1, maxValueState.value())
                }
            }
        )
        res.print()
        env.execute()

    }
}
